#include "zrpc/base/async_log.h"

namespace zrpc {
AsyncLogging::AsyncLogging(std::string file_path, std::string base_name,
                           size_t roll_size, int32_t interval_)
    : file_path_(file_path),
      base_name_(base_name),
      interval_(interval_),
      running_(false),
      roll_size_(roll_size),
      current_buffer_(new Buffer),
      next_buffer_(new Buffer),
      buffers_() {
  current_buffer_->Bzero();
  next_buffer_->Bzero();
  buffers_.reserve(16);
}

void AsyncLogging::Append(const char *logline, size_t len) {
  std::unique_lock<std::mutex> lk(mutex_);
  if (current_buffer_->Avail() > len) {
    current_buffer_->Append(logline, len);
  } else {
    buffers_.push_back(std::unique_ptr<Buffer>(current_buffer_.release()));
    if (next_buffer_) {
      current_buffer_ = std::move(next_buffer_);
    } else {
      current_buffer_.reset(new Buffer);
    }
    current_buffer_->Append(logline, len);
    condition_.notify_one();
  }
}

void AsyncLogging::ThreadFunc() {
  LogFile output(file_path_, base_name_, roll_size_, false);
  std::unique_ptr<Buffer> new_buffer1(new Buffer);
  std::unique_ptr<Buffer> new_buffer2(new Buffer);
  new_buffer1->Bzero();
  new_buffer2->Bzero();

  BufferVector buffers_to_write;
  buffers_to_write.reserve(16);

  running_ = true;

  while (running_) {
    assert(new_buffer1 && new_buffer1->Length() == 0);
    assert(new_buffer2 && new_buffer2->Length() == 0);
    assert(buffers_to_write.empty());
    {
      std::unique_lock<std::mutex> lk(mutex_);
      if (buffers_.empty()) {
        condition_.wait_for(lk, std::chrono::seconds(interval_));
      }

      buffers_.push_back(std::unique_ptr<Buffer>(current_buffer_.release()));
      current_buffer_ = std::move(new_buffer1);
      buffers_to_write.swap(buffers_);

      if (!next_buffer_) {
        next_buffer_ = std::move(new_buffer2);
      }
    }

    assert(!buffers_to_write.empty());

    if (buffers_to_write.size() > 25) {
      char buf[256];
      snprintf(buf, sizeof buf,
               "Dropped log messages at %s, %zd larger buffers_\n",
               TimeStamp::Now().ToFormattedString().c_str(),
               buffers_to_write.size() - 2);
      fputs(buf, stderr);
      output.Append(buf, static_cast<int>(::strlen(buf)));
      buffers_to_write.erase(buffers_to_write.begin() + 2,
                             buffers_to_write.end());
    }

    for (size_t i = 0; i < buffers_to_write.size(); ++i) {
      output.Append(buffers_to_write[i]->GetData(),
                    buffers_to_write[i]->Length());
    }

    if (buffers_to_write.size() > 2) {
      buffers_to_write.resize(2);
    }

    if (!new_buffer1) {
      assert(!buffers_to_write.empty());
      new_buffer1 = std::move(buffers_to_write.back());
      buffers_to_write.pop_back();
      new_buffer1->Reset();
    }

    if (!new_buffer2) {
      assert(!buffers_to_write.empty());
      new_buffer2 = std::move(buffers_to_write.back());
      buffers_to_write.pop_back();
      new_buffer2->Reset();
    }

    buffers_to_write.clear();
    output.Flush();
  }
  output.Flush();
}

}  // namespace zrpc